package com.icehand.canal.client.transfer;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import com.icehand.canal.annotation.HandlePoint;
import com.icehand.canal.client.CanalClientException;
import com.icehand.canal.client.Point;
import com.icehand.canal.config.CanalConfig;
import com.icehand.canal.handler.CanalHandler;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Method;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.Predicate;

/**
 * @author icehand
 */
@Slf4j
public abstract class AbstractMessageTransponder extends BaseMessageTransponder{
    public AbstractMessageTransponder(CanalConnector canalConnector, CanalConfig.Instance instance, String destination,
                                      List<CanalHandler> handlers, List<Point> points) {
        super(canalConnector, instance, destination, handlers, points);
    }

    @Override
    protected void distributeMessage(Message message) {
        List<CanalEntry.Entry> entries = message.getEntries();
        for (CanalEntry.Entry entry : entries) {
            List<CanalEntry.EntryType> ignoreEntryTypes = getIgnoreEntryTypes();
            if (null != ignoreEntryTypes && ignoreEntryTypes.stream().anyMatch(t -> entry.getEntryType() == t)) {
                continue;
            }

            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (InvalidProtocolBufferException e) {
                throw new CanalClientException("ERROR,parse row change error,"+entry.toString(),e);
            }

            if(rowChange.hasIsDdl() && rowChange.getIsDdl()) {
                //DDL
                processDDL(rowChange);
                continue;
            }
            for(CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                distributeByImpl(rowChange.getEventType(),rowData);
                distributeByAnnotation(destination, entry.getHeader().getSchemaName(),
                        entry.getHeader().getTableName(),rowChange.getEventType(),rowData);
            }

        }
    }

    protected abstract void processDDL(CanalEntry.RowChange rowChange);

    protected void distributeByImpl(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
        if(null != handlers) {
            handlers.forEach(handler -> handler.handle(eventType,rowData));
        }
    }

    protected void distributeByAnnotation(String destination, String schemaName, String tableName,
                                          CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
        points.forEach(point -> {
            point.getInvokeMap().entrySet().stream().filter(getAnnotationFilter(destination, schemaName, tableName, eventType))
                    .forEach(entry -> {
                        Method method = entry.getKey();
                        method.setAccessible(true);
                        Object[] args = getInvokeArgs(method,eventType, rowData);
                        try {
                            method.invoke(point.getTarget(),args);
                        } catch (Exception e) {
                            log.error("{}: Error occurred when invoke the listener's interface! class:{}, method:{}",
                                    Thread.currentThread().getName(),
                                    point.getTarget().getClass().getName(), method.getName());
                        }
                    });
        });
    }

    protected List<CanalEntry.EntryType> getIgnoreEntryTypes() {
        return Collections.emptyList();
    }

    protected abstract Predicate<Map.Entry<Method, HandlePoint>> getAnnotationFilter(String destination,
                                                                                     String schemaName,
                                                                                     String tableName,
                                                                                     CanalEntry.EventType eventType);

    protected abstract Object[] getInvokeArgs(Method method, CanalEntry.EventType eventType,
                                              CanalEntry.RowData rowData);
}
